package mq.config;

import mq.listener.MQListenerCurrently;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import static org.apache.rocketmq.common.consumer.ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET;

/**
 * @Author:sjb
 * @CreateTime:2024-08-23
 * @Description: 生产者消费者初始化
 * @Version：1.0
 */
@Configuration
public class MqConfig {

    private static final Logger log = LoggerFactory.getLogger(MqConfig.class);
    @Value("${mq.producer.nameServer:127.0.0.1:9876}")
    private String nameServer;
    @Value("${mq.producer.maxMessageSize:4096}")
    private Integer maxMessageSize;
    @Value("${mq.producer.groupName:TRAINING_TOPIC_MQ}")
    private String groupName;
    @Value("${mq.producer.timeOut:sendMsgTimeOut:30000}")
    private Integer sendMsgTimeOut;
    @Value("${mq.producer.retryWhenSendFailed:2}")
    private Integer retryWhenSendFailed;

    @Value("${mq.consumer.groupName:TRAINING_TOPIC_CONSUMER}")
    private String consumerGroupName;
    @Value("${mq.consumer.minThread:20}")
    private int minThread;
    @Value("${mq.consumer.maxThread:60}")
    private int maxThread;
    @Value("${mq.consumer.topic:TRAINING_TOPIC_MQ}")
    private String topic;


    @Bean("default")
    public DefaultMQProducer getDefaultProducer() {
        DefaultMQProducer producer = new DefaultMQProducer(this.groupName);
        producer.setMaxMessageSize(this.maxMessageSize);
        producer.setNamesrvAddr(this.nameServer);
        producer.setSendMsgTimeout(this.sendMsgTimeOut);
        producer.setRetryTimesWhenSendFailed(this.retryWhenSendFailed);

        try {
            producer.start();
        } catch (MQClientException e) {
            log.error("init mq producer error:{}", e.getMessage());
        }
        return producer;
    }


    @Bean
    public DefaultMQPushConsumer getDefaultConsumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.consumerGroupName);
        consumer.setNamesrvAddr(this.nameServer);
        consumer.setConsumeThreadMin(this.minThread);
        consumer.setConsumeThreadMax(this.maxThread);
        consumer.setMessageListener(getMessageListenerConcurrently());
        consumer.setConsumeFromWhere(CONSUME_FROM_FIRST_OFFSET);

        try {
            consumer.subscribe(topic,"*");
            consumer.start();
        } catch (MQClientException e) {
            log.error("init mq consumer error:{}", e.getMessage());
        }
        return consumer;
    }


    @Bean
    public MessageListenerConcurrently getMessageListenerConcurrently() {
        return new MQListenerCurrently();
    }
}
